package com.yc.distribute.lock.zookeeper;

import java.util.Random;
import java.util.concurrent.CountDownLatch;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.KeeperException.NodeExistsException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooDefs;

import com.yc.distribute.lock.DistributedLock;

/**
 * zookeeper实现分布式锁
 *
 */
public class ZookeeperLock implements DistributedLock {
	private String LOCK_PATH;//竞争的锁
	private static ZooKeeper zk;
	private String data;
	CountDownLatch cdl = null;
	
	/**
	 * 实际开发中，连接会在项目启动时就初始化好zookeeper
	 * @throws Exception
	 */
	public ZookeeperLock(String lockPath) throws Exception {
		LOCK_PATH = lockPath;
		cdl = new CountDownLatch(1);
		zk = new ZooKeeper("172.16.67.127:2181,172.16.67.128:2181,172.16.67.122:2181", 5000, new Watcher() {
			
			@Override
			public void process(WatchedEvent event) {
				System.out.println("事件：节点" + event.getPath() + " - 类型：" + event.getType());
				if (KeeperState.SyncConnected == event.getState()) {//连接成功
					if (EventType.None == event.getType() && null == event.getPath()) {
						System.out.println("连接成功");
						cdl.countDown();
					}
					
				}
				
			}
		});
		
		//zookeeper异步处理。使用线程门栓同步等待,直到连接成功,再返回
		cdl.await();
	}

	@Override
	public void lock() throws Exception {
		if (!tryLock()) {
			waitLock();
		}

	}
	
	private void waitLock() throws Exception {
		cdl = new CountDownLatch(1);
		//监听锁（LOCK_PATH）
		zk.getData(LOCK_PATH, new Watcher() {
			
			@Override
			public void process(WatchedEvent event) {
				try {
					if (EventType.NodeDeleted == event.getType() && tryLock()) {//关键：节点被删除时就会触发 尝试抢锁
						cdl.countDown();
					}
				} catch (Exception e) {
					e.printStackTrace();
				}
				
			}
		}, new Stat());
		
		//在自己获得锁之前一直等待
		cdl.await();
	}

	@Override
	public boolean tryLock() throws Exception {
		try {
			data = System.currentTimeMillis() + "" + new Random().nextInt(99999999);//唯一标识自己的数据
			zk.create(LOCK_PATH, data.getBytes("UTF-8"), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
		} catch (NodeExistsException e) {
			return false;
		}
		return true;
	}

	@Override
	public void unlock() {
		if (null != zk) {
			try {
				zk.delete(LOCK_PATH, -1);//-1不管版本号，直接删除
			} catch (KeeperException e) {
				e.printStackTrace();
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}

	}

}
